package com.rxdemo;

import androidx.appcompat.app.AppCompatActivity;

import android.annotation.SuppressLint;
import android.content.Context;
import android.content.SharedPreferences;
import android.net.NetworkRequest;
import android.net.Uri;
import android.os.Bundle;
import android.text.TextUtils;
import android.util.Log;
import android.view.View;

import com.google.gson.Gson;

import java.util.ArrayList;
import java.util.List;
import java.util.Random;
import java.util.concurrent.Callable;
import java.util.concurrent.TimeUnit;

import io.reactivex.Flowable;
import io.reactivex.Observable;
import io.reactivex.ObservableEmitter;
import io.reactivex.ObservableOnSubscribe;
import io.reactivex.ObservableSource;
import io.reactivex.Observer;
import io.reactivex.Single;
import io.reactivex.SingleObserver;
import io.reactivex.android.schedulers.AndroidSchedulers;
import io.reactivex.disposables.Disposable;
import io.reactivex.functions.BiFunction;
import io.reactivex.functions.Consumer;
import io.reactivex.functions.Function;
import io.reactivex.functions.Predicate;
import io.reactivex.schedulers.Schedulers;
import okhttp3.Call;
import okhttp3.OkHttpClient;
import okhttp3.Request;
import okhttp3.Response;
import okhttp3.ResponseBody;
// TODO: 2019/10/9 此项目的缘由就是为了练习RxJava

public class MainActivity extends AppCompatActivity {
    private static final String TAG = "MainActivity";

    private StringBuilder stringBuilder;
    private SharedPreferences sharedPreferences;

    boolean saveTestCache = false;
    private String key = "test";
    private Disposable disposable;

    @Override
    protected void onCreate(Bundle savedInstanceState) {
        super.onCreate(savedInstanceState);
        setContentView(R.layout.activity_main);
        stringBuilder = new StringBuilder();

        sharedPreferences = getSharedPreferences("rxDemo", MODE_PRIVATE);
    }

    public void onClick(View view) {
        switch (view.getId()) {
            case R.id.btn_create:
                initCreate();
                break;
            case R.id.btn_map:
                initMap();
                break;
            case R.id.btn_zip:
                initZip();
                break;
            case R.id.btn_concat:
                initConcat();
                break;
            case R.id.btn_flat_map:
                initFlatMap();
                break;
            case R.id.btn_concat_map:
                initConcatMap();
                break;
            case R.id.btn_distinct:
                initDistinct();
                break;
            case R.id.btn_filter:
                initFilter();
                break;
            case R.id.btn_buffer:
                initBuffer();
                break;
            case R.id.btn_timer:
                initTimer();
                break;
            case R.id.btn_interval:
                initInterval();
                break;
            case R.id.btn_do_on_next:
                initDoOnNext();
                break;
            case R.id.btn_skip:
                initSkip();
                break;
            case R.id.btn_take:
                initTake();
                break;
            case R.id.btn_just:
                initJust();
                break;
            case R.id.btn_single:
                initSingle();
                break;
            case R.id.btn_debounce:
                initDebounce();
                break;
            case R.id.btn_defer:
                initDefer();
                break;
            case R.id.btn_last:
                initLast();
                break;
            case R.id.btn_merge:
                initMerge();
                break;
            case R.id.btn_reduce:
                initReduce();
                break;
            case R.id.btn_scan:
                initScan();
                break;
            case R.id.btn_window:
                initWindow();
                break;
            case R.id.btn_network:
                initNetWork();
                break;
            case R.id.btn_cache:
                initCache();
                break;
            case R.id.btn_many_request:
                initManyRequest();
                break;
            case R.id.btn_zip_net:
                initZipNet();
                break;
            case R.id.btn_interval_net:
                initIntervalNet();
                break;

        }
    }

    private void initIntervalNet() {
        disposable = Flowable.interval(1, TimeUnit.SECONDS)
                .doOnNext(new Consumer<Long>() {
                    @Override
                    public void accept(Long aLong) throws Exception {
                        Log.e(TAG, "accept: " + aLong);
                    }
                }).observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Consumer<Long>() {
                    @Override
                    public void accept(Long aLong) throws Exception {
                        Log.e(TAG, "accept: 设置文本 => " + aLong);
                    }
                });
    }

    /**
     * 整合多个网络数据到一个处理器中
     */
    private void initZipNet() {
        Observable<BannerBean> beanObservable = Observable.create(new ObservableOnSubscribe<BannerBean>() {
            @Override
            public void subscribe(ObservableEmitter<BannerBean> emitter) throws Exception {
                Request request = new Request.Builder()
                        .url("https://wanandroid.com/article/listproject/0/json")
                        .get()
                        .build();
                Response response = new OkHttpClient().newCall(request).execute();
                if (response.isSuccessful()) {
                    Gson gson = new Gson();
                    BannerBean bannerBean = gson.fromJson(response.body().string(), BannerBean.class);
                    emitter.onNext(bannerBean);
                }
            }
        });

        Observable<WXBean> wxBeanObservable = Observable.create(new ObservableOnSubscribe<WXBean>() {
            @Override
            public void subscribe(ObservableEmitter<WXBean> emitter) throws Exception {
                Request.Builder builder = new Request.Builder()
                        .url("https://wanandroid.com/wxarticle/chapters/json")
                        .get();
                Request request = builder.build();
                Call newCall = new OkHttpClient().newCall(request);
                Response response = newCall.execute();
                if (response.isSuccessful()) {
                    ResponseBody responseBody = response.body();
                    if (responseBody != null) {
                        String json = responseBody.string();
                        Log.e(TAG, "apply: 转换前 +>" + json);

                        WXBean wxBean = new Gson().fromJson(json, WXBean.class);
                        emitter.onNext(wxBean);
                        sharedPreferences.edit().putString(key, wxBean.toString()).apply();
                    }
                } else {
                    Log.e(TAG, "subscribe: 请求失败");
                    emitter.onComplete();
                }
            }
        });

        Disposable disposable = Observable.zip(wxBeanObservable, beanObservable, new BiFunction<WXBean, BannerBean, String>() {
            @Override
            public String apply(WXBean wxBean, BannerBean bannerBean) throws Exception {
                String name = wxBean.getData().get(0).getName();
                String author = bannerBean.getData().getDatas().get(0).getAuthor();
                return "wxBean => " + name + " author => " + author;
            }
        }).subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Consumer<String>() {
                    @Override
                    public void accept(String s) throws Exception {
                        Log.e(TAG, "accept: " + s);
                    }
                });
    }

    /**
     * 多个网络依次请求
     */
    private void initManyRequest() {

        Disposable disposable = Observable.create(new ObservableOnSubscribe<WXBean>() {
            @Override
            public void subscribe(ObservableEmitter<WXBean> emitter) throws Exception {
                Log.e(TAG, "subscribe: 第一次请求 开始睡眠");
                Thread.sleep(5000);
                Request.Builder builder = new Request.Builder()
                        .url("https://wanandroid.com/wxarticle/chapters/json")
                        .get();
                Request request = builder.build();
                Call newCall = new OkHttpClient().newCall(request);
                Response response = newCall.execute();
                if (response.isSuccessful()) {
                    String string = response.body().string();
                    Gson gson = new Gson();
                    WXBean wxBean = gson.fromJson(string, WXBean.class);
                    emitter.onNext(wxBean);
                    Log.e(TAG, "subscribe: 第一次开始请求");
                }

            }
        }).subscribeOn(Schedulers.io())
                .observeOn(Schedulers.io())
                .flatMap(new Function<WXBean, ObservableSource<WXBean>>() {
                    @Override
                    public ObservableSource<WXBean> apply(WXBean wxBean) throws Exception {

                        return Observable.create(new ObservableOnSubscribe<WXBean>() {
                            @Override
                            public void subscribe(ObservableEmitter<WXBean> emitter) throws Exception {
                                Log.e(TAG, "subscribe: 第二次请求");
                                Request.Builder builder = new Request.Builder()
                                        .url("https://wanandroid.com/wxarticle/chapters/json")
                                        .get();
                                Request request = builder.build();
                                Call newCall = new OkHttpClient().newCall(request);
                                Response response = newCall.execute();
                                if (response.isSuccessful()) {
                                    String string = response.body().string();
                                    Gson gson = new Gson();
                                    WXBean wxBean = gson.fromJson(string, WXBean.class);
                                    emitter.onNext(wxBean);
                                    Log.e(TAG, "subscribe: 第二次开始请求");
                                }
                            }
                        });
                    }
                }).subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Consumer<WXBean>() {
                    @Override
                    public void accept(WXBean wxBean) throws Exception {
                        Log.e(TAG, "accept: 已经接受到请求了 " + wxBean.toString());
                    }
                });
    }

    /**
     * 先读取缓存然后再进行网络访问
     */
    private void initCache() {
        Observable<WXBean> cacheObservable = Observable.create(new ObservableOnSubscribe<WXBean>() {
            @Override
            public void subscribe(ObservableEmitter<WXBean> emitter) throws Exception {
                Log.e(TAG, "subscribe: 当前读取线程 => " + Thread.currentThread().getName());

                String cache = sharedPreferences.getString(key, "");
                if (TextUtils.isEmpty(cache)) {
                    Log.e(TAG, "subscribe: 开启网络请求");
                    emitter.onComplete();
                } else {
                    Gson gson = new Gson();
                    WXBean wxBean = gson.fromJson(cache, WXBean.class);
                    emitter.onNext(wxBean);
                    Log.e(TAG, "subscribe: 缓存中读取到了数据");
                }
            }
        });

        Observable<WXBean> netObservable = Observable.create(new ObservableOnSubscribe<WXBean>() {
            @Override
            public void subscribe(ObservableEmitter<WXBean> emitter) throws Exception {
                Request.Builder builder = new Request.Builder()
                        .url("https://wanandroid.com/wxarticle/chapters/json")
                        .get();
                Request request = builder.build();
                Call newCall = new OkHttpClient().newCall(request);
                Response response = newCall.execute();
                if (response.isSuccessful()) {
                    ResponseBody responseBody = response.body();
                    if (responseBody != null) {
                        String json = responseBody.string();
                        Log.e(TAG, "apply: 转换前 +>" + json);

                        WXBean wxBean = new Gson().fromJson(json, WXBean.class);
                        emitter.onNext(wxBean);
                        sharedPreferences.edit().putString(key, wxBean.toString()).apply();
                    }
                } else {
                    Log.e(TAG, "subscribe: 请求失败");
                    emitter.onComplete();
                }
            }
        });

        Disposable disposable = Observable.concat(cacheObservable, netObservable)
                .subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Consumer<WXBean>() {
                    @Override
                    public void accept(WXBean wxBean) throws Exception {
                        Log.e(TAG, "accept: 接收到了 => " + wxBean.toString());
                    }
                }, new Consumer<Throwable>() {
                    @Override
                    public void accept(Throwable throwable) throws Exception {
                        Log.e(TAG, "accept: 读取失败=> " + throwable.getMessage());
                    }
                });
    }

    /**
     * 基本网络请求
     */
    private void initNetWork() {
        Disposable disposable = Observable.create(new ObservableOnSubscribe<Response>() {
            @Override
            public void subscribe(ObservableEmitter<Response> emitter) throws Exception {
                Request.Builder builder = new Request.Builder()
                        .url("https://wanandroid.com/wxarticle/chapters/json")
                        .get();
                Request request = builder.build();
                Call newCall = new OkHttpClient().newCall(request);
                Response response = newCall.execute();
                emitter.onNext(response);
            }
        }).map(new Function<Response, WXBean>() {
            @Override
            public WXBean apply(Response response) throws Exception {
                Log.e(TAG, "apply: " + Thread.currentThread().getName());
                if (response.isSuccessful()) {
                    ResponseBody responseBody = response.body();
                    if (responseBody != null) {

                        String json = responseBody.string();
                        Log.e(TAG, "apply: 转换前 +>" + json);

                        return new Gson().fromJson(json, WXBean.class);
                    }
                }
                return null;
            }
        }).observeOn(AndroidSchedulers.mainThread())
                .doOnNext(new Consumer<WXBean>() {
                    @Override
                    public void accept(WXBean wxBean) throws Exception {
                        Log.e(TAG, "accept: " + Thread.currentThread().getName());
                        Log.e(TAG, "accept: " + wxBean.toString());
                    }
                }).subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Consumer<WXBean>() {
                    @Override
                    public void accept(WXBean wxBean) throws Exception {
                        Log.e(TAG, "accept: " + Thread.currentThread().getName());
                        Log.e(TAG, "accept: 成功了=> " + wxBean.toString());

                    }
                }, new Consumer<Throwable>() {
                    @Override
                    public void accept(Throwable throwable) throws Exception {
                        Log.e(TAG, "accept: " + Thread.currentThread().getName());
                        Log.e(TAG, "accept: 失败了=> " + throwable.getMessage());
                    }
                });
    }

    private void initWindow() {
        Disposable subscribe = Observable.interval(1, TimeUnit.SECONDS)
                .take(15)
                .window(3, TimeUnit.SECONDS)
                .subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Consumer<Observable<Long>>() {
                    @Override
                    public void accept(Observable<Long> longObservable) throws Exception {
                        Log.e(TAG, "accept: window => 开始");
                        Disposable disposable = longObservable.subscribeOn(Schedulers.io())
                                .observeOn(AndroidSchedulers.mainThread())
                                .subscribe(new Consumer<Long>() {
                                    @Override
                                    public void accept(Long aLong) throws Exception {
                                        Log.e(TAG, "accept: window=> " + aLong);
                                    }
                                });
                    }
                });
    }

    /**
     * 和reduce类似 但是它会输出每一个步骤而reduce 只会输出最终结果
     */
    private void initScan() {
        Disposable disposable = Observable.just(1, 2, 3)
                .scan(new BiFunction<Integer, Integer, Integer>() {
                    @Override
                    public Integer apply(Integer integer, Integer integer2) throws Exception {
                        return integer + integer2;
                    }
                }).subscribe(new Consumer<Integer>() {
                    @Override
                    public void accept(Integer integer) throws Exception {
                        Log.e(TAG, "accept: reduce=>" + integer);
                    }
                });
    }

    /**
     * 每次使用一个方法处理一个值,有一个初始值
     */
    private void initReduce() {
        Disposable disposable = Observable.just(1, 2, 3)
                .reduce(new BiFunction<Integer, Integer, Integer>() {
                    @Override
                    public Integer apply(Integer integer, Integer integer2) throws Exception {
                        return integer + integer2;
                    }
                }).subscribe(new Consumer<Integer>() {
                    @Override
                    public void accept(Integer integer) throws Exception {
                        Log.e(TAG, "accept: reduce=>" + integer);
                    }
                });
    }

    /**
     * 不同于concat,把多个observable 结合起来一起发送,不用等A发射器发射玩再发射B发射器
     */
    private void initMerge() {
        Disposable subscribe = Observable.merge(Observable.just(1, 2, 3, 4567, 89, 90, 0, 0, 6, 4), Observable.just(4, 5))
                .subscribe(new Consumer<Integer>() {
                    @Override
                    public void accept(Integer integer) throws Exception {
                        Log.e(TAG, "accept: merge=>" + integer);
                    }
                });
    }

    /**
     * 取出可观察到的最后一个值,或者满足某些条件的最后一项
     */
    private void initLast() {
        Disposable subscribe = Observable.just(1, 2, 3, 8, 9, 5, 3)
                .last(4)
                .subscribe(new Consumer<Integer>() {
                    @Override
                    public void accept(Integer integer) throws Exception {
                        Log.e(TAG, "accept: last: " + integer);
                    }
                });
    }

    /**
     * 每次订阅都会创建一个新的Observable,如果没有订阅就不会产生
     */
    private void initDefer() {
        Observable<Integer> defer = Observable.defer(new Callable<ObservableSource<Integer>>() {
            @Override
            public ObservableSource<Integer> call() throws Exception {
                return Observable.just(1, 2, 3);
            }
        });
        defer.subscribe(new Observer<Integer>() {
            @Override
            public void onSubscribe(Disposable d) {

            }

            @Override
            public void onNext(Integer integer) {
                Log.e(TAG, "onNext: defer : " + integer);
            }

            @Override
            public void onError(Throwable e) {

            }

            @Override
            public void onComplete() {

            }
        });
    }

    /**
     * 去除发送频率过快的项 小于500ms 都不予以接收
     * 两个相邻数据发射的时间间隔决定了前一个数据是否要丢弃
     */
    private void initDebounce() {
        Disposable subscribe = Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
                emitter.onNext(1);
                Thread.sleep(400);
                emitter.onNext(2);
                Thread.sleep(505);
                emitter.onNext(3);
                Thread.sleep(100);
                emitter.onNext(4);
                Thread.sleep(605);
                emitter.onNext(5);
                Thread.sleep(510);
                emitter.onComplete();
            }
        }).debounce(500, TimeUnit.MILLISECONDS)
                .subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Consumer<Integer>() {
                    @Override
                    public void accept(Integer integer) throws Exception {
                        Log.e(TAG, "accept: debounce" + integer);
                    }
                });
    }

    /**
     * 只接受一个参数 只调用 onSuccess/onError
     */
    private void initSingle() {

        Single.just(new Random().nextInt())
                .subscribe(new SingleObserver<Integer>() {
                    @Override
                    public void onSubscribe(Disposable d) {

                    }

                    @Override
                    public void onSuccess(Integer integer) {
                        Log.e(TAG, "onSuccess: " + integer);
                    }

                    @Override
                    public void onError(Throwable e) {
                        Log.e(TAG, "onError: " + e.getMessage());
                    }
                });
    }

    /**
     * 一个简单的发射器,依次调用onNext
     */
    private void initJust() {
        Disposable subscribe = Observable.just("123", "456", "789")
                .subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Consumer<String>() {
                    @Override
                    public void accept(String s) throws Exception {
                        Log.e(TAG, "accept: just " + s);
                    }
                });
    }

    private void initTake() {
        Disposable subscribe = Observable.just(1, 2, 3, 4, 5)
                .take(2)
                .subscribe(new Consumer<Integer>() {
                    @Override
                    public void accept(Integer integer) throws Exception {
                        Log.e(TAG, "accept: take: " + integer);
                    }
                });
    }

    /**
     * 表示跳过几个事件然后开始接收
     */
    private void initSkip() {
        Disposable disposable = Observable.just(1, 2, 3, 4, 5)
                .skip(2)
                .subscribe(new Consumer<Integer>() {
                    @Override
                    public void accept(Integer integer) throws Exception {
                        Log.e(TAG, "accept: skip: " + integer);
                    }
                });
    }

    /**
     * 在接收到事件之前做一些操作
     */
    private void initDoOnNext() {
        Disposable disposable = Observable.just(1, 2, 3, 4)
                .doOnNext(new Consumer<Integer>() {
                    @Override
                    public void accept(Integer integer) throws Exception {
                        Log.e(TAG, "accept: doOnNext 保存成功");
                    }
                }).subscribe(new Consumer<Integer>() {
                    @Override
                    public void accept(Integer integer) throws Exception {

                        Log.e(TAG, "accept: doOnNext 接收成功");
                    }
                });
    }

    /**
     * 定时事件
     */
    private void initInterval() {
        Disposable subscribe = Observable.interval(3, 2, TimeUnit.SECONDS)
                .subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Consumer<Long>() {
                    @Override
                    public void accept(Long aLong) throws Exception {
                        Log.e(TAG, "interval accept: " + TimeUnit.SECONDS.name());

                    }
                });
    }

    /**
     * 延迟发送事件
     */
    //timer默认在新线程,需要切换回主线程
    private void initTimer() {
        Log.e(TAG, "initTimer: timer start" + TimeUnit.SECONDS);
        Disposable subscribe = Observable.timer(2, TimeUnit.SECONDS)
                .subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Consumer<Long>() {
                    @Override
                    public void accept(Long aLong) throws Exception {
                        Log.e(TAG, "initTimer: accept: " + TimeUnit.SECONDS);
                    }
                });
    }

    /**
     * count 是最大取值 ,skip跳过事件
     * 把一系列的事件分批次传送过来
     */
    private void initBuffer() {
        Disposable disposable = Observable.just(1, 2, 3, 4, 5)
                .buffer(3, 2)
                .subscribe(new Consumer<List<Integer>>() {
                    @Override
                    public void accept(List<Integer> integers) throws Exception {
                        Log.e(TAG, "buffer accept: 接收数据");
                        for (int i :
                                integers) {
                            Log.e(TAG, "buffer accept: " + i);
                        }
                    }
                });
    }

    /**
     * 条件判断事件,过滤不符合条件的事件
     */
    private void initFilter() {
        Disposable disposable = Observable.just(1, 20, 65, -5, 7, 19)
                .filter(new Predicate<Integer>() {
                    @Override
                    public boolean test(Integer integer) throws Exception {
                        return integer >= 10;
                    }
                }).subscribe(new Consumer<Integer>() {
                    @Override
                    public void accept(Integer integer) throws Exception {
                        Log.e(TAG, "filter  accept: " + integer);
                    }
                });
    }

    /**
     * 会过滤掉重复的事件
     */
    private void initDistinct() {
        Disposable subscribe = Observable.just(1, 1, 1, 2, 2, 3, 4, 5)
                .distinct()
                .subscribe(new Consumer<Integer>() {
                    @Override
                    public void accept(Integer integer) throws Exception {
                        Log.e(TAG, "distinct  accept: " + integer);
                    }
                });
    }

    /**
     * 与flatMap 类似 但是它是有序的
     */
    private void initConcatMap() {
        Disposable subscribe = Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
                emitter.onNext(1);
                emitter.onNext(2);
                emitter.onNext(3);
            }
        }).concatMap(new Function<Integer, ObservableSource<String>>() {
            @Override
            public ObservableSource<String> apply(Integer integer) throws Exception {
                ArrayList<String> strings = new ArrayList<>();
                for (int i = 0; i < 3; i++) {
                    strings.add("I am value: " + integer);
                }
                int delayTime = (int) (1 + Math.random() * 10);
                return Observable.fromIterable(strings).delay(delayTime, TimeUnit.MILLISECONDS);
            }
        }).subscribeOn(Schedulers.newThread())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Consumer<String>() {
                    @Override
                    public void accept(String s) throws Exception {
                        Log.e(TAG, "flatMap  accept: " + s);
                    }
                });
    }

    /**
     * 它可以把一个发射器 Observable 通过某种方法转换为多个 Observables，
     * 然后再把这些分散的 Observables装进一个单一的发射器 Observable。
     * 但有个需要注意的是，flatMap 并不能保证事件的顺序
     */

    private void initFlatMap() {
        Disposable subscribe = Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
                emitter.onNext(1);
                emitter.onNext(2);
                emitter.onNext(3);
            }
        }).flatMap(new Function<Integer, ObservableSource<String>>() {
            @Override
            public ObservableSource<String> apply(Integer integer) throws Exception {

                List<String> list = new ArrayList<>();
                for (int i = 1; i < 4; i++) {
                    list.add("I am value " + integer);
                    Log.e(TAG, "apply: " + i);
                }
                int delayTime = (int) (1 + Math.random() * 10);
                Log.e(TAG, "apply: delayTime => " + delayTime);
                return Observable.fromIterable(list).delay(delayTime, TimeUnit.MILLISECONDS);
            }
        }).subscribeOn(Schedulers.newThread())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Consumer<String>() {
                    @Override
                    public void accept(String s) throws Exception {
                        System.out.println(s);
                        Log.e(TAG, "flatMap  accept: " + s);
                    }
                }, new Consumer<Throwable>() {
                    @Override
                    public void accept(Throwable throwable) throws Exception {
                        Log.e(TAG, "accept: flatMap异常: " + throwable.getMessage());
                    }
                });
    }

    /**
     * 通过concat 把两个或者多个合并成一个发射器,顺序发送
     * 只有前一个发射完毕,调用了emitter.onComplete(),下一个才会发送
     */
    private void initConcat() {
        Disposable disposable = Observable.concat(Observable.just(1, 2, 3), Observable.just(4, 5, 6))
                .subscribe(new Consumer<Integer>() {
                    @Override
                    public void accept(Integer integer) throws Exception {
                        Log.e(TAG, "accept: concat " + integer);
                    }
                });
    }

    /**
     * 通过zip,来合并两个发射器的数据,以最少的为准,
     */
    private void initZip() {
        Disposable subscribe = Observable.zip(getStringObservable(), getIntegerObservable(), new BiFunction<String, Integer, String>() {
            @Override
            public String apply(String s, Integer integer) throws Exception {
                return s + integer;
            }
        }).subscribe(new Consumer<String>() {
            @Override
            public void accept(String s) throws Exception {
                Log.e(TAG, "accept: " + s);
            }
        });
    }

    /**
     * 通过map函数来操作发射器的数据 转换成其他格式的数据
     */
    private void initMap() {
        Disposable subscribe = Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
                emitter.onNext(1);
                emitter.onNext(2);
                emitter.onNext(3);
            }
        }).map(new Function<Integer, String>() {
            @Override
            public String apply(Integer integer) throws Exception {
                //进行一种转换的模式
                return "the result is +> " + integer;
            }
        }).subscribe(new Consumer<String>() {
            @Override
            public void accept(String s) throws Exception {
                Log.e(TAG, "accept: 转换之后接收+> " + s);
            }
        });

    }

    /**
     * 最基本的创建方式
     */
    private void initCreate() {
        Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
                stringBuilder.append("发射器 emit 1  \n");
                Log.e(TAG, "subscribe: 发射器 emit  1 \n");
                emitter.onNext(1);
                stringBuilder.append("发射器 emit 2  \n");
                Log.e(TAG, "subscribe: 发射器 emit  2 \n");
                emitter.onNext(2);

                stringBuilder.append("发射器 emit 3  \n");
                Log.e(TAG, "subscribe: 发射器 emit  3 \n");
                emitter.onNext(3);
                emitter.onComplete();

                stringBuilder.append("发射器 emit 4  \n");
                Log.e(TAG, "subscribe: 发射器 emit  4 \n");
                emitter.onNext(4);

            }
        }).subscribe(new Observer<Integer>() {

            private int i;
            private Disposable mDisposable;

            @Override
            public void onSubscribe(Disposable d) {
                stringBuilder.append("接收器: ").append(d.isDisposed()).append(" \n");
                Log.e(TAG, "onSubscribe: " + "接收器: " + d.isDisposed() + " \n");
                mDisposable = d;
            }

            @Override
            public void onNext(Integer integer) {
                stringBuilder.append(" 接收器: onNext: value: ").append(integer);
                Log.e(TAG, " 接收器: onNext: value: " + integer);
                i++;
                if (i == 2) {
                    mDisposable.dispose();  //可以直接切断接收器
                    stringBuilder.append(" 接收器: onNext:isDisposable ").append(mDisposable.isDisposed());
                    Log.e(TAG, " 接收器: onNext:isDisposable " + mDisposable.isDisposed());
                }
            }

            @Override
            public void onError(Throwable e) {
                stringBuilder.append(" 接收器: onError: value: ").append(e.getMessage());
                Log.e(TAG, "接收器: onError: value: " + e.getMessage());
            }

            @Override
            public void onComplete() {
                stringBuilder.append(" 接收器: onComplete:");
                Log.e(TAG, "接收器: onComplete: ");
            }
        });
    }

    private Observable<String> getStringObservable() {
        return Observable.create(new ObservableOnSubscribe<String>() {
            @Override
            public void subscribe(ObservableEmitter<String> emitter) throws Exception {
                if (!emitter.isDisposed()) {
                    emitter.onNext("A");
                    Log.e(TAG, "subscribe: 发射器: A");
                    emitter.onNext("B");
                    Log.e(TAG, "subscribe: 发射器: B");
                    emitter.onNext("C");
                    Log.e(TAG, "subscribe: 发射器: C");
                }
            }
        });
    }

    private Observable<Integer> getIntegerObservable() {
        return Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
                if (!emitter.isDisposed()) {
                    emitter.onNext(1);
                    Log.e(TAG, "subscribe: 发射器: 1");
                    emitter.onNext(2);
                    Log.e(TAG, "subscribe: 发射器: 2");
                    emitter.onNext(3);
                    Log.e(TAG, "subscribe: 发射器: 3");
                    emitter.onNext(4);
                    Log.e(TAG, "subscribe: 发射器: 4");
                    emitter.onNext(5);
                    Log.e(TAG, "subscribe: 发射器: 5");
                }
            }
        });
    }


    @Override
    protected void onDestroy() {
        super.onDestroy();
        if (disposable != null && !disposable.isDisposed()) {
            disposable.dispose();
        }
    }
}
